package cn.edu.ccnu.nlp;


import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.util.Version;

public class IndexingReplace 
{
	
	public static Configuration configuration;
	public static HTablePool pool;
	public static HTable news_index;
	public static HTable news_stat;
	static {
		configuration = HBaseConfiguration.create();
		
		try{
			pool = new HTablePool(configuration, 1000);
			news_index = (HTable)pool.getTable("news_index");
			news_stat = (HTable)pool.getTable("news_stat");
		}catch(Exception e){
			e.getStackTrace();
		}
	}

	public static SimpleDateFormat formatter_time = new SimpleDateFormat("yyyy-MM-dd HH.mm.ss");

	/*public static void main(String args[])  throws Exception
	{
		process("news");
	}
	public static void process(String corpus)  throws Exception
	{
			Class.forName("com.mysql.jdbc.Driver").newInstance();
			Connection con1=(Connection) java.sql.DriverManager.getConnection("jdbc:mysql://202.114.40.175/corpus?useUnicode=true&characterEncoding=UTF-8","root","ccnunlp2010@nlp.com");
			Statement stmt=(Statement) con1.createStatement();
			
			Directory directory = FSDirectory.open(new File("txt/"+corpus + "_1"));

			IndexWriter text_writer = new IndexWriter(directory, new StandardAnalyzer(Version.LUCENE_30), true,IndexWriter.MaxFieldLength.UNLIMITED);
			text_writer.setMergeFactor(100);
			text_writer.setMaxBufferedDocs(100);

			int start = 1;	
			int size = 1000;

			while(true)
			{
				ResultSet rst = stmt.executeQuery("select id, title,date from " +  corpus + " where id >= " + start + " and  id < " + (start + size));
				int n_count = 0;
				while (rst.next())
				{
					try
					{
						n_count++;

						String date = rst.getString("date").replaceAll("-", "");				
						String title = rst.getString("title");
						String id = rst.getString("id");

						Document text_doc = new Document();
						text_doc.add(new Field("title",  title, Field.Store.YES, Field.Index.ANALYZED ));
						text_doc.add(new Field("id", id, Field.Store.YES, Field.Index.NO ));
						text_doc.add(new Field("date", date, Field.Store.YES, Field.Index.NOT_ANALYZED  ));
						text_writer.addDocument(text_doc);
					}
					catch (Exception e) {
						e.printStackTrace(System.out);
					}	
				}
				rst.close();

				System.out.println(start);

				if( n_count == 0)
					break;

				start += size;
			}
				
			con1.close();
			text_writer.optimize();
			text_writer.close();
	}*/

	/**
	 * 从hbase的表读取数据并创建表的索引
	 * 
	 */
	public void process(String corpus) throws CorruptIndexException, IOException{
		
		SimpleDateFormat sdf=new SimpleDateFormat("yyyy");                     //获取当前年份，以便于遍历2005年到当前的年份
		String current_year=sdf.format(new java.util.Date());  
		String[] months = {"01","02","03","04","05","06","07","08","09","10","11","12"};
		
		/**
		 * 设置索引存放路径
		 */
		Directory directory = FSDirectory.open(new File("txt/"+corpus + "_1"));
		IndexWriter text_writer = new IndexWriter(directory, new StandardAnalyzer(Version.LUCENE_30), true,IndexWriter.MaxFieldLength.UNLIMITED);
		text_writer.setMergeFactor(100);
		text_writer.setMaxBufferedDocs(100);
		
		/**
		 * 都取数据并建索引
		 * 按一个月一个月来读
		 */
		ResultScanner rs = null;
		int totalRows = 0;
		String year;
		try{
			for(int i = 2005; i < Integer.parseInt(current_year); i++){
				year = ""+i;
				for(int j = 0; j < months.length; j++){
					
					Filter filter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(year+"-"+months[j]+"-.*")); // 查询某年某月的记录数
					Scan scan = new Scan();
					scan.setFilter(filter);
				
					rs = news_index.getScanner(scan);
					Result result;
					
					while((result = rs.next()) != null){
						//System.out.println((new String(result.getRow())));
						for(KeyValue keyValue:result.raw()){
						String date = (new String(keyValue.getRow())).replaceAll("-", "");
						String url = new String(keyValue.getQualifier());
						String title = (new String(keyValue.getValue())).split(",")[1];
						
						//建索引
						Document text_doc = new Document();
						text_doc.add(new Field("title",  title, Field.Store.YES, Field.Index.ANALYZED ));
						text_doc.add(new Field("url", url, Field.Store.YES, Field.Index.NO ));
						text_doc.add(new Field("date", date, Field.Store.YES, Field.Index.NOT_ANALYZED  ));
						text_writer.addDocument(text_doc);
	//					System.out.println("date:"+ date+":url:"+url+":title:"+title);
						totalRows++;
						}
						System.out.println("totalRow:"+totalRows);
					}
				}
			}
			}catch(Exception e){
				e.getStackTrace();
			}finally{
				rs.close();
			}
		text_writer.optimize();
		text_writer.close();
		System.out.println("共读取了"+totalRows+"条数据。");
		System.out.println(corpus+"数据库表的索引创建完成！");
	}

	public static void main(String[] args) throws CorruptIndexException, IOException{
		IndexingReplace ir = new IndexingReplace();
		ir.process("news");
	}


}